﻿
using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Diagnostics;

namespace WSU.ChekaParallel.TestTask
{
    static class TestContinuation
    {
        private static void ContinueAndWait(Task startingTask)
        {
            // chekanote: by default, TPL never guarantee that the continuation will run on the same thread as the antecedant task
            Task followingTask = startingTask.ContinueWith(ant =>
            {
                Console.WriteLine(
                    "status of antTask is '{0}', continuation runs on thread<{1}>",
                    ant.Status, Thread.CurrentThread.ManagedThreadId);
                if (ant.Exception != null)
                    Console.WriteLine("!!! Antecedent's Exception is : {0}", ant.Exception.InnerException.Message);
            });
            startingTask.Start();
            followingTask.Wait();
            Console.WriteLine();
        }

        private static void TestSimpleContinueNormal()
        {
            // ===================== continue with normal task
            Task normalTask = new Task(() => Console.WriteLine("normal task starts at thread<{0}>", Thread.CurrentThread.ManagedThreadId));
            ContinueAndWait(normalTask);
        }

        private static void TestSimpleContinueCanceled()
        {
            CancellationTokenSource cancelSource = new CancellationTokenSource();
            CancellationToken cancelToken = cancelSource.Token;
            Task cancelableTask = new Task(() =>
            {
                Console.WriteLine("cancelable task starts at thread<{0}>", Thread.CurrentThread.ManagedThreadId);
                while (true)
                {
                    cancelToken.ThrowIfCancellationRequested();
                    Console.Write("*");
                    Thread.Sleep(100);
                    Console.Write("=");
                }
            }, cancelToken);

            Task.Factory.StartNew(() =>
            {
                Thread.Sleep(1000);
                cancelSource.Cancel();
            });
            ContinueAndWait(cancelableTask);
        }

        private static void TestSimpleContinueException()
        {
            Task exceptionTask = new Task(() =>
            {
                Console.WriteLine("exception task starts on thread<{0}>", Thread.CurrentThread.ManagedThreadId);
                Thread.Sleep(100);
                throw new InvalidOperationException();
            });
            ContinueAndWait(exceptionTask);
        }

        private static void TestRethrowException()
        {
            Action throwException = () => { throw new InvalidOperationException { Source = "Antecedent Task" }; };

            // ----------------- continuation doesn't re-thrown
            // then waiting on continuation will not kill the whole application
            Task contNoThrowTask = Task.Factory.StartNew(throwException).ContinueWith(anteTask => Console.WriteLine("No re-throwing"));
            contNoThrowTask.Wait(); // will not throw any exception and kill the whole application

            // ----------------- if you want to let the caller know that an exception has already happened
            // note: you should explicitly check the Exception property of the antecedent task and re-throw the original exception
            // chekanote: otherwise, waiting on a continuation will not throw the exception from the original starting task
            Task contRethrowTask = Task.Factory.StartNew(throwException).ContinueWith(anteTask =>
            {
                if (anteTask.Exception != null)
                {
                    Debug.Assert(anteTask.Exception is AggregateException);
                    throw anteTask.Exception.InnerException;
                }
                Debug.Fail("this line will never be executed.");
            });

            try
            {
                contRethrowTask.Wait();
                Debug.Fail("this line will never be executed");
            }
            catch (AggregateException aexp)
            {
                Console.WriteLine("{0}: {1}", aexp.InnerException.Source, aexp.InnerException.Message);
            }
        }

        private static Task[] ContinueTasksFactory(bool excepEnabled)
        {
            Task anteTask = Task.Factory.StartNew(state =>
            {
                if ((bool)state)
                {
                    Console.WriteLine("antecedent task throws exception.");
                    throw new InvalidOperationException();
                }
                else
                {
                    Console.WriteLine("antecedent task completes successfully.");
                }
            }, excepEnabled);

            Task okContTask = anteTask.ContinueWith(ant => Console.WriteLine("no exception occurs at previous task"), TaskContinuationOptions.OnlyOnRanToCompletion);
            Task excepContTask = anteTask.ContinueWith(ant => Console.WriteLine("previous task throws exception: {0}", ant.Exception.InnerException.Message), TaskContinuationOptions.OnlyOnFaulted);

            return new Task[] { okContTask, excepContTask };
        }

        /// <summary>
        /// several continuations linked to the same antecedent task
        /// only one of them will be fired, and all the others will be canceled
        /// chekanote: so you cannot use WaitAny to wait for that really executed continuation, 
        /// because called continuations finish much earlier
        /// and if you use WaitAll, you should pay attention that "OperationCanceledException" will be thrown
        /// </summary>
        /// <param name="allContiTasks">all continuations following the same antecedent task</param>
        private static void WaitAllContinuePath(this Task[] allContiTasks)
        {
            try
            {
                Task.WaitAll(allContiTasks);
                // chekanote: below codes will never be executed because one of the 'allContiTasks' will be canceled
                // when okContinue and failContinue both continue on the same antecedent 
                // whatever that antecedent returns, both paths, "ok" and "fail" will be scheduled,
                // one of them will run, and the other will be canceled, none of them will be forgotten
                Debug.Fail("never execute this line due to previous exception.");
            }
            catch (AggregateException aexp)
            {
                Debug.Assert(aexp.InnerExceptions.Count == allContiTasks.Length - 1,
                    "one of the path will run, the others will be canceled");

                foreach (Exception innerException in aexp.InnerExceptions)
                {
                    Debug.Assert(innerException is OperationCanceledException);
                }
            }
        }

        private static void TestConditionContinue()
        {
            ContinueTasksFactory(true).WaitAllContinuePath();
            ContinueTasksFactory(false).WaitAllContinuePath();
        }

        private static void TestContinueWithChildExceptions()
        {
            Task parentTask = Task.Factory.StartNew(() =>
            {
                Task.Factory.StartNew(() => { throw new NotImplementedException { Source = "ChildTask1" }; }, TaskCreationOptions.AttachedToParent);
                Task.Factory.StartNew(() => { throw new InvalidOperationException { Source = "ChildTask2" }; }, TaskCreationOptions.AttachedToParent);
                Task.Factory.StartNew(() => Console.WriteLine("<ChildTask3> ends"), TaskCreationOptions.AttachedToParent);
            });

            Task continueTask = parentTask.ContinueWith(ant =>
            {
                foreach (Exception innerExcep in ant.Exception.InnerExceptions)
                {
                    Console.WriteLine("exception<{0}>: {1}", innerExcep.InnerException.Source, innerExcep.InnerException.Message);
                }
            });

            continueTask.Wait();
        }

        /// <summary>
        /// when multiple continuations linked with the same starting task
        /// if these continuations are exclusive, then only one of them will be finally invoked
        /// chekanote: but those not-invoked ones are not forgotten or abandon, they are just canceled
        /// which means those not-invoked ones are still completed, if any other tasks are linked to those not-invoked ones
        /// they will be fired
        /// </summary>
        private static void TestContiuePathStatus()
        {
            Action<Task[], int> checkContiStatus = (contiTasks, canceledIndex) =>
            {
                contiTasks.WaitAllContinuePath();
                Debug.Assert(contiTasks.All(task => task.IsCompleted));
                Debug.Assert(contiTasks[canceledIndex].IsCanceled);
            };

            checkContiStatus(ContinueTasksFactory(excepEnabled: true), 0);// the OK continuation path is canceled
            checkContiStatus(ContinueTasksFactory(excepEnabled: false), 1);// the error continuation path is canceled
        }

        private static void TestContinueWhenAll_ExplicitType()
        {
            Func<int, int, Task<int>> factory = (sleepTime, result) => Task.Factory.StartNew(() =>
                                                                                                 {
                                                                                                     Thread.Sleep(sleepTime);
                                                                                                     return result;
                                                                                                 });
            Task<int>[] tasks = new Task<int>[]
            {
                factory(100,50),
                factory(500,20),
                Task.Factory.StartNew(()=> 
                { 
                    throw new InvalidOperationException();
                    return 99;
                })
            };

            Task<float> compositeTask = Task<float>.Factory.ContinueWhenAll<int>(tasks, antecedents =>
            {
                int total = 0;
                foreach (Task<int> anteTask in antecedents)
                {
                    try
                    {
                        total += anteTask.Result;
                    }
                    catch (AggregateException aggregatedError)
                    {
                        Console.WriteLine("exception caught");
                        aggregatedError.Flatten().Handle(ex => ex is InvalidOperationException);
                    }
                }// foreach

                return total * 1.5f;
            });

            Console.WriteLine("final composite result={0}", compositeTask.Result);
        }

        /// <summary>
        /// test the ContinueWhenAll, each antecedent has different type
        /// and the task got from ContinueWhenAll has another type
        /// </summary>
        private static void TestContiueWhenAllWithResult()
        {
            Task<int> intTask = Task.Factory.StartNew<int>(() =>
                                                               {
                                                                   Thread.Sleep(100);
                                                                   Console.WriteLine("int task done.");
                                                                   return 1;
                                                               });
            Task<string> strTask = Task.Factory.StartNew(() =>
                                                             {
                                                                 Thread.Sleep(500);
                                                                 Console.WriteLine("string task done.");
                                                                 return "cheka";
                                                             });
            Task<float> aggregateTask = Task.Factory.ContinueWhenAll(new Task[] { intTask, strTask }, tasks =>
            {
                int intresult = (tasks[0] as Task<int>).Result;
                string strresult = (tasks[1] as Task<string>).Result;
                return intresult + strresult.Length * 0.1f;
            });
            Console.WriteLine("aggregated result = {0}", aggregateTask.Result);
        }

        private static void TestContinueAny()
        {
            CancellationTokenSource cts = new CancellationTokenSource();
            CancellationToken cancelToken = cts.Token;

            Random random = new Random();
            Task<int>[] tasks = new Task<int>[3];
            for (int index = 0; index < tasks.Length; index++)
            {
                int sleepInterval = random.Next(200, 5000);
                int taskNo = index;
                tasks[index] = new Task<int>(() =>
                                                 {
                                                     cancelToken.WaitHandle.WaitOne(sleepInterval);
                                                     if (cancelToken.IsCancellationRequested)
                                                     {
                                                         Console.WriteLine("!!! task<{0}> cancelled", taskNo);
                                                         throw new OperationCanceledException(cancelToken);
                                                     }
                                                     else
                                                     {
                                                         Console.WriteLine("task<{0}> returns {1}", taskNo, sleepInterval);
                                                         return sleepInterval;
                                                     }
                                                 }, cancelToken);
            }

            Task<float> compositeTask = Task.Factory.ContinueWhenAny(tasks,
                firstFinished => firstFinished.Result * 100.0f);

            foreach (Task<int> task in tasks)
            {
                task.Start();
            }
            Console.WriteLine("final result={0}", compositeTask.Result);

            // speculative processing, when we have already got the first result
            // cancel others to save resource
            cts.Cancel();
            try
            {
                Task.WaitAll(tasks);
            }
            catch (AggregateException aggregateError)
            {
                aggregateError.Flatten().Handle(ex => ex is OperationCanceledException);
            }

            Debug.Assert(tasks.All(t => t.IsCompleted));
            Console.WriteLine("DONE !!!");
        }

        private static void TestSyncContinue()
        {
            Action anteAct = () =>
                                 {
                                     Thread.Sleep(1000);
                                     Console.WriteLine("antecedent runs on thread<{0}>", Thread.CurrentThread.ManagedThreadId);
                                 };
            Action<Task> contiAct = ante => Console.WriteLine("continuation runs on thread<{0}>", Thread.CurrentThread.ManagedThreadId);

            // -------------- default continue
            Console.WriteLine("Default, TPL never guarantee that continuation runs on the same thread with antecedent.");
            Task.Factory.StartNew(anteAct).ContinueWith(contiAct).Wait();

            // -------------- specify continue synchronously
            Console.WriteLine("\nSpecify continuation runs on the same thread as the antecedent.");
            Task.Factory.StartNew(anteAct).ContinueWith(contiAct, TaskContinuationOptions.ExecuteSynchronously).Wait();
        }

        private static void TestContinueWithCancel()
        {
            CancellationTokenSource cts = new CancellationTokenSource();

            Task sourceTask = Task.Factory.StartNew(() =>
                                                        {
                                                            Console.WriteLine("source task is running,......");
                                                            cts.Token.WaitHandle.WaitOne();
                                                            cts.Token.ThrowIfCancellationRequested();
                                                        }, cts.Token);
            // since this continuation is constructed with the same CancelToken
            // so when the antecedent is canceled, the token must already been canceled, so it will never be scheduled
            Task neverScheduled = sourceTask.ContinueWith(
                antecedent => Debug.Fail("Never be scheduled"),
                cts.Token,
                TaskContinuationOptions.OnlyOnCanceled,
                TaskScheduler.Current);

            Task willContinueTask = sourceTask.ContinueWith(
                antecedent => Console.WriteLine("Continuation runs"), TaskContinuationOptions.OnlyOnCanceled);

            Task noneContinueTask = sourceTask.ContinueWith(
                antecedent => Debug.Fail("Never be executed"), TaskContinuationOptions.NotOnCanceled);

            Helper.Pause();
            cts.Cancel();

            try
            {
                Task.WaitAll(neverScheduled, willContinueTask, noneContinueTask);
            }
            catch (AggregateException aggregatedError)
            {
                aggregatedError.Flatten().Handle(ex => ex is OperationCanceledException);
            }

            Debug.Assert(neverScheduled.IsCompleted);
            Debug.Assert(neverScheduled.IsCanceled);

            Debug.Assert(willContinueTask.IsCompleted);
            Debug.Assert(willContinueTask.Status == TaskStatus.RanToCompletion);

            Debug.Assert(noneContinueTask.IsCompleted);
            Debug.Assert(noneContinueTask.IsCanceled);
            Debug.Assert(noneContinueTask.Status == TaskStatus.Canceled);
        }

        /// <summary>
        /// chekanote: The NotOn* and OnlyOn* TaskContinuationOptions, 
        /// which constrain for which TaskStatus states a continuation will be executed, are illegal with ContinueWhenAny.
        /// </summary>
        private static void TestContinueAny_ForbidOptions()
        {
            Task oktask = Task.Factory.StartNew(() => Console.WriteLine("ok."));
            Task errortask = Task.Factory.StartNew(() => { throw new InvalidOperationException("test"); });

            try
            {
                Task<int> continuation = Task.Factory.ContinueWhenAny(new Task[] { oktask, errortask },
                                                                  firstCompleted => 0,
                                                                  TaskContinuationOptions.OnlyOnRanToCompletion);
                Debug.Fail("Impossible to execute this line of code");
            }
            catch (ArgumentOutOfRangeException ex)
            {
                Console.WriteLine("!!!!!!!!!!!!!! Exception Caught: \n" + ex.Message);
            }
        }

        public static void TestMain()
        {
            // TestSimpleContinueNormal();
            // TestSimpleContinueException();
            // TestSimpleContinueCanceled();
            // TestRethrowException();
            // TestConditionContinue();
            // TestContinueWithChildExceptions();
            // TestContiuePathStatus();
            // TestContiueWhenAllWithResult();
            // TestSyncContinue();
            // TestContinueWhenAll_ExplicitType();
            // TestContinueAny();
            // TestContinueWithCancel();
            TestContinueAny_ForbidOptions();
        }
    }
}